package my.thesis.bolts.aggregators;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import my.thesis.analytics.metrics.ThroughputMetric;
import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class MentionsCounter extends BaseRichBolt
{
	private ConcurrentHashMap<String, HashMap<String, Integer>> userMentionsFollowers;
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private int tiToEmit;
	
	private Boolean receivedSinceLastEmit;
	transient ThroughputMetric _throughputMetric;
	private static Boolean metricsOn=false;
	
	public MentionsCounter(int tiSec)
	{
		this.tiToEmit=tiSec;
	}
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		userMentionsFollowers= new ConcurrentHashMap<String, HashMap<String, Integer>>();
		this.collector=collector;
		
		this.receivedSinceLastEmit=false;
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("mentionCount", _throughputMetric, 20);		
			_throughputMetric.setStartTime();
		}
	}
	public void execute(Tuple tuple) 
	{
		if(isTickTuple(tuple))
		{			
			if(this.receivedSinceLastEmit)
			{
				//TEST
				System.out.println("Emitting mentions...");
				
				emitMentions();
			}
			
			this.receivedSinceLastEmit=false;
		}
		else
		{	
			this.receivedSinceLastEmit=true;
			
			if(metricsOn)
			{
				_throughputMetric.incrTuples();
			}
			
			String username = tuple.getStringByField("username");
			String mention = tuple.getStringByField("mention");
			int followers= tuple.getIntegerByField("followers");
			
			HashMap<String, Integer> mapForUserFollowers = getMapForUserFollowers(username);
			mapForUserFollowers.put(mention, followers);
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("userMentionsFollowers"));
	}
	
    private HashMap<String, Integer> getMapForUserFollowers(String username) 
    {
    	try
    	{
    	  HashMap<String, Integer> mapForUser = userMentionsFollowers.get(username);
		    
		  if (mapForUser == null) 
		  {
			  	mapForUser=new HashMap<String, Integer>();
			  	userMentionsFollowers.put(username, mapForUser);
		    	return userMentionsFollowers.get(username);
		  }
		  else 
		  { 
			  return userMentionsFollowers.get(username);
		  }
    	}
    	catch(Exception ex)
    	{
    		System.out.println(ex.toString());
    		return null;
    	}
	}
        
    @Override
	 public Map<String, Object> getComponentConfiguration() 
	 {
	    Config conf = new Config();
	    conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, this.tiToEmit);
	    return conf;
	 }
	 
    private boolean isTickTuple(Tuple tuple)
	{
		 String sourceComponent = tuple.getSourceComponent();
		 String sourceStreamId = tuple.getSourceStreamId();
		 return sourceComponent.equals(Constants.SYSTEM_COMPONENT_ID) 
					 && sourceStreamId.equals(Constants.SYSTEM_TICK_STREAM_ID);
	} 
    
	private void emitMentions()
	{	
		 ConcurrentHashMap<String, HashMap<String, Integer>> results = new ConcurrentHashMap<String, HashMap<String, Integer>>(userMentionsFollowers);
		
		 collector.emit(new Values(results));

		 userMentionsFollowers.clear();
	}
}
